package api.flinkapi.apidemos;

import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.Iterator;

/***
 * @author hunter.d
 * @qq 657270652
 * @wx haitao-duan
 * @date 2021/7/25
 **/
public class ListStateDemo {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setParallelism(1);
        DataStreamSource<String> ds = env.socketTextStream("localhost", 5656);
        KeyedStream<String, String> keyd = ds.keyBy(s -> s.split(",")[0]);
        SingleOutputStreamOperator<String> res = keyd.process(new KeyedProcessFunction<String, String, String>() {
            ListState<String> listState;

            @Override
            public void open(Configuration parameters) throws Exception {

                listState = getRuntimeContext().getListState(new ListStateDescriptor<String>("lst", String.class));
            }

            @Override
            public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
                listState.add(value);

                StringBuilder sb = new StringBuilder();
                Iterator<String> iterator = listState.get().iterator();
                while (iterator.hasNext()) {
                    sb.append(iterator.next()).append("-");
                }

                sb.append("=========");
                while (iterator.hasNext()) {
                    sb.append(iterator.next()).append("-");
                }

                out.collect(sb.toString());
            }
        });

        res.print();
        env.execute();


    }

}
